package com.example.mqtt.utils;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;

/**
 * 这里仅添加几个用到的方法，用到后再添加
 * */
@Slf4j
public class MqttClientUtils extends MqttClient{
    protected static final Map<String,MqttClientUtils> CLIENT_UTILS_MAP = new ConcurrentHashMap<>();

    public MqttClientUtils(String broker,String clientId) throws MqttException {
        super(broker,clientId);
    }

    public MqttClientUtils(String broker,String clientId, MqttClientPersistence persistence) throws MqttException {
        super(broker,clientId,persistence);
    }

    public MqttClientUtils(String broker,String clientId, MqttClientPersistence persistence, ScheduledExecutorService executorService) throws MqttException {
        super(broker,clientId,persistence,executorService);
    }


    public static MqttClientUtils getInstance(String serverURI,String clientId){
        try {
            if (!CLIENT_UTILS_MAP.containsKey(serverURI)) {
                MqttClientUtils mqttClientUtils = new MqttClientUtils(serverURI, clientId);
                CLIENT_UTILS_MAP.put(serverURI,mqttClientUtils);
            }
        } catch (MqttException e) {
            log.error("mqtt创建异常",e);
            throw new RuntimeException(e);
        }
        return CLIENT_UTILS_MAP.get(serverURI);
    }

    public void connect(){
        this.connect(new MqttConnectOptions());
    }

    public void connect(MqttConnectOptions options) {
        try {
            super.connect(options);
        } catch (MqttException e) {
            log.error("mqtt连接异常",e);
            throw new RuntimeException(e);
        }
    }

    public IMqttToken connectWithResult(MqttConnectOptions options){
        try {
            return super.connectWithResult(options);
        } catch (MqttException e) {
            log.error("mqtt连接异常",e);
            throw new RuntimeException(e);
        }
    }

    public void disconnect() {
        try {
            super.disconnect();
            CLIENT_UTILS_MAP.remove(getServerURI());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    public void disconnect(long quiesceTimeout) {
        try {
            super.disconnect(quiesceTimeout);
            CLIENT_UTILS_MAP.remove(getServerURI());
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) throws Exception {
        String broker = "tcp://183.220.28.163:11883"; // MQTT服务器地址
        // 设置连接选项
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setUserName("test");
        options.setPassword("test".toCharArray());
        // 创建MqttClient对象并连接到MQTT服务器
        MqttClient client = new MqttClient(broker, MqttClient.generateClientId());
        client.connect(options);

        // 订阅主题
        String topic = "thing/product/+/osd";
        client.subscribe(topic, (tc,message)->{
            System.out.println(tc);
            byte[] payload = message.getPayload();
            System.out.println(new String(payload));
        });

        // 订阅主题
        String topic1 = "thing/product/+/state";
        client.subscribe(topic1, (tc,message)->{
            System.out.println(tc);
            byte[] payload = message.getPayload();
            System.out.println(new String(payload));
        });

        System.out.println("已经订阅了主题：" + topic);
        // 发布消息
        String message = "Hello, MQTT!";
        byte[] payloadBytes = message.getBytes();
        MqttMessage messageToSend = new MqttMessage(payloadBytes);
        int qosLevel = 2; // QoS等级为2
        messageToSend.setQos(qosLevel);
        boolean retained = false; // 不保留消息
        messageToSend.setRetained(retained);
        long expiryInterval = 60 * 1000L; // 消息过期时间为60秒
        // 这个版本没有？
//        messageToSend.setExpiryInterval(expiryInterval);
        client.publish("/test/topic", messageToSend);
        System.out.println("已发送消息：" + message);

        // 断开与MQTT服务器的连接
        client.disconnect();
        System.out.println("已断开与MQTT服务器的连接");
    }
}
